package com.gudy.counter.bean;

import com.google.common.collect.ImmutableMap;
import com.gudy.counter.config.CounterConfig;
import com.gudy.counter.util.DbUtil;
import com.gudy.counter.util.IDConverter;
import com.gudy.counter.util.JsonUtil;
import io.netty.util.collection.IntObjectHashMap;
import io.netty.util.collection.LongObjectHashMap;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.ArrayUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import thirdpart.hq.L1MarketData;
import thirdpart.hq.MatchData;
import thirdpart.order.OrderCmd;
import thirdpart.order.OrderDirection;
import thirdpart.order.OrderStatus;

import javax.annotation.PostConstruct;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.gudy.counter.bean.MqttBusConsumer.INNER_MARKET_DATA_CACHE_ADDR;
import static com.gudy.counter.bean.MqttBusConsumer.INNER_MATCH_DATA_ADDR;
import static com.gudy.counter.config.WebSocketConfig.ORDER_NOTIFY_ADDR_PREFIX;
import static com.gudy.counter.config.WebSocketConfig.TRADE_NOTIFY_ADDR_PREFIX;

/**
 * 接收撮合交易数据
 * 对变动的数据做业务处理
 * 给订阅的终端发送通知
 */
@Log4j2
@Component
public class MatchDataConsumer {

    public static final String ORDER_DATA_CACHE_ADDR = "order_data_cache_addr";

    @Autowired
    private CounterConfig config;

    /*委托编号-> OrderCmd */
    private LongObjectHashMap<OrderCmd> oidMapOrder = new LongObjectHashMap<>();


    @PostConstruct
    private void init(){
        EventBus eventBus = config.getVertx().eventBus();

        //接收委托缓存 从终端来
        eventBus.consumer(ORDER_DATA_CACHE_ADDR).handler(this::handlerOrderDataCacheAddr);

        eventBus.consumer(INNER_MATCH_DATA_ADDR).handler(
                buffer->{
                    Buffer body = (Buffer)buffer.body();
                    if(body.length()==0) return;

                    MatchData[] matchDataArr = null;
                    try {
                        matchDataArr = config.getBodyCodec().deserialize(body.getBytes(), MatchData[].class);
                    } catch (Exception e) {
                        log.error(e);
                    }
                    if(ArrayUtils.isEmpty(matchDataArr)) return;

                    Map<Long, List<MatchData>> oidMapMatchdatas = Stream.of(matchDataArr).collect(Collectors.groupingBy(matchData -> matchData.oid));
                    for (Map.Entry<Long, List<MatchData>> entry : oidMapMatchdatas.entrySet()) {
                        if(CollectionUtils.isEmpty(entry.getValue())) return;

                        Long oid = entry.getKey();
                        int counterOId = IDConverter.seperateLong2Int(oid)[1];
                        updateAndNotify(counterOId,entry.getValue(),oidMapOrder.get(oid));
                    }
                }
        );

    }

    private void updateAndNotify(int counterOId, List<MatchData> value, OrderCmd orderCmd) {
        if(CollectionUtils.isEmpty(value)) return;

        //成交数据  数据变动存到数据库
        for (MatchData md : value) {
            OrderStatus status = md.status;
            if(status == OrderStatus.TRADE_ED || status == OrderStatus.PART_TRADE){
                DbUtil.saveTrade(counterOId,md,orderCmd);

                if(orderCmd.direction == OrderDirection.BUY){
                    /*委托价格高于实际成交价格时 需要释放资金*/
                    if(orderCmd.price > md.price){
                        DbUtil.addBalance(orderCmd.uid,(orderCmd.price-md.price) * md.volume);
                    }
                    DbUtil.addPosi(orderCmd.uid,orderCmd.code,md.volume,md.price);
                }else if(orderCmd.direction == OrderDirection.SELL){
                    DbUtil.addBalance(orderCmd.uid, md.price * md.volume);
                }else{
                    log.error("wrong direction[{}]", orderCmd.direction);
                }
                /*通知客户端*/
                config.getVertx().eventBus().publish(
                        TRADE_NOTIFY_ADDR_PREFIX + orderCmd.uid,
                        JsonUtil.toJson(
                                ImmutableMap.of("code",orderCmd.code,
                                        "direction",orderCmd.direction,
                                        "volume",md.volume)
                        )
                );
            }
        }

        /*处理最后一笔委托*/
        MatchData finalMatchData = value.get(value.size() - 1);
        OrderStatus finalOrderStatus = finalMatchData.status;
        DbUtil.updateOrder(orderCmd.uid,counterOId, finalOrderStatus);
        if(finalOrderStatus == OrderStatus.CANCEL_ED || finalOrderStatus == OrderStatus.PART_CANCEL){
            oidMapOrder.remove(orderCmd.oid);
            if(orderCmd.direction == OrderDirection.BUY){
                DbUtil.addBalance(orderCmd.uid,-(orderCmd.price*finalMatchData.volume));
            }else if(orderCmd.direction == OrderDirection.SELL){
                DbUtil.addPosi(orderCmd.uid,orderCmd.code,-finalMatchData.volume,orderCmd.price);
            }else{
                log.error("wrong direction[{}]", orderCmd.direction);
            }
        }

        config.getVertx().eventBus().publish(
                ORDER_NOTIFY_ADDR_PREFIX + orderCmd.uid,""
        );

    }

    private void handlerOrderDataCacheAddr(Message<Object> buffer) {
        Buffer body = (Buffer)buffer.body();
        try {
            OrderCmd cmd = config.getBodyCodec().deserialize(body.getBytes(), OrderCmd.class);
            log.info("cache order:{}",cmd);
            oidMapOrder.put(cmd.oid,cmd);
        } catch (Exception e) {
            log.error(e);
        }
    }
}
